package com.xiashitech.apiservice.grpc;

import cn.hutool.json.JSONUtil;
import com.xiashitech.apiservice.api.ApiMetadataController;
import com.xiashitech.interfaces.agent.dto.api.ApiRequest;
import com.xiashitech.interfaces.grpc.*;
import com.xiashitech.interfaces.grpc.exchange.CommunicationObject;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
//@Service
@GrpcService
public class ApiInfoService extends RPCApiInfoServiceGrpc.RPCApiInfoServiceImplBase {
    private final ConcurrentHashMap<String, CommunicationObject<String>> apiReceivedData = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, CommunicationObject<String>> desenseReceivedData = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, CommunicationObject<String>> paramsReceivedData = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    public ApiInfoService() {
        scheduledExecutorService.scheduleAtFixedRate(()->{
            // 清理接收数据的缓存
            // log.info(" now apiReceivedData size: " + apiReceivedData.size());
            Iterator<Map.Entry<String, CommunicationObject<String>>> iterator = apiReceivedData.entrySet().iterator();
            while(iterator.hasNext()) {
                Map.Entry<String, CommunicationObject<String>> item = iterator.next();
                long timespan = System.currentTimeMillis() - item.getValue().getSendTimeStamp();
                if(timespan > 24 * 60 * 60* 1000 + 10 * 60 * 1000) { // 24 小时 + 10 分钟后过期
                    iterator.remove();
                }
            }

            // log.info(" now desenseReceivedData size: " + desenseReceivedData.size());
            iterator = desenseReceivedData.entrySet().iterator();
            while(iterator.hasNext()) {
                Map.Entry<String, CommunicationObject<String>> item = iterator.next();
                long timespan = System.currentTimeMillis() - item.getValue().getSendTimeStamp();
                if(timespan > 24 * 60 * 60* 1000 + 10 * 60 * 1000) { // 24 小时 + 10 分钟后过期
                    iterator.remove();
                }
            }

            // log.info(" now paramsReceivedData size: " + paramsReceivedData.size());
            iterator = paramsReceivedData.entrySet().iterator();
            while(iterator.hasNext()) {
                Map.Entry<String, CommunicationObject<String>> item = iterator.next();
                long timespan = System.currentTimeMillis() - item.getValue().getSendTimeStamp();
                if(timespan > 24 * 60 * 60* 1000 + 10 * 60 * 1000) { // 24 小时 + 10 分钟后过期
                    iterator.remove();
                }
            }
        }, 30, 10, TimeUnit.MINUTES);
    }

    @Override
    public StreamObserver<RPCApiSyncRequest> grpcApiToOmdStream(StreamObserver<RPCApiSyncResponse> responseObserver) {
        StreamObserver<RPCApiSyncRequest> requestStreamObserver = new StreamObserver<RPCApiSyncRequest> (){
            public void onNext(RPCApiSyncRequest value) {
                // 处理收到的请求，不做响应
                String apisStr = value.getApisJson();
                log.info(Thread.currentThread().getId() + ", grpcApiToOmdStream onNext enter ...., apisStr string length: " + apisStr.length());
                if(apisStr == null || "".equals(apisStr))
                    return;
                // 回传收到的 requestUuid
                String reqUuid = value.getRequestUuid();
                RPCApiSyncResponse respObject = RPCApiSyncResponse.newBuilder().setCode("OK").setRequestUuid(reqUuid).build();
                responseObserver.onNext(respObject);
                responseObserver.onCompleted();
                // 构建本地缓存，避免重传对业务的影响
                if(apiReceivedData.get(reqUuid) != null) { // 假如已存在，返回，不提交到业务端
                    log.info(Thread.currentThread().getId() + ", grpcApiToOmdStream onNext reqUuid " + reqUuid + " has EXISTED, skip");
                    return;
                }
                CommunicationObject<String> co = new CommunicationObject<String>();
                co.setSendTimeStamp(System.currentTimeMillis());
                co.setExchangeUuid(reqUuid);
                co.setData("");
                apiReceivedData.put(reqUuid, co);

                ApiRequest apiRequest = JSONUtil.toBean(apisStr, ApiRequest.class);
                ApiMetadataController.submitTask(apiRequest, "syncApisToOpenMetadata"); // 静态线程池成员，不会多实例，也线程安全
                log.info(Thread.currentThread().getId() + ", grpcApiToOmdStream onNext end, has onNext/onComplete ...." + System.identityHashCode(responseObserver));
            }

            @Override
            public void onError(Throwable t) {
                log.info(Thread.currentThread().getId() + ", grpcApiToOmdStream onError : " + t.getMessage());
            }

            @Override
            public void onCompleted() {
                log.info(Thread.currentThread().getId() + ", grpcApiToOmdStream onCompleted " + System.identityHashCode(responseObserver));
            }

        };
        return requestStreamObserver;
    }

    @Override
    public StreamObserver<RPCDesenseParamRequest> biSyncApiDesense(StreamObserver<RPCDesenseParamResponse> responseObserver) {
        StreamObserver<RPCDesenseParamRequest> requestStreamObserver = new StreamObserver<RPCDesenseParamRequest> (){
            public void onNext(RPCDesenseParamRequest value) {
                // 处理收到的请求，不做响应
                String dpStr = value.getDesenseParamJson();
                log.info(Thread.currentThread().getId() + ", biSyncApiDesense onNext enter ...., dpStr string length: " + dpStr.length());
                if(dpStr == null || "".equals(dpStr))
                    return;
                // 回传收到的 requestUuid
                String reqUuid = value.getRequestUuid();
                RPCDesenseParamResponse rpcResponse = RPCDesenseParamResponse.newBuilder().setCode("OK").setRequestUuid(reqUuid).build();
                responseObserver.onNext(rpcResponse);
                responseObserver.onCompleted();
                // 构建本地缓存，避免重传对业务的影响
                if(desenseReceivedData.get(reqUuid) != null) { // 假如已存在，返回，不提交到业务端
                    log.info(Thread.currentThread().getId() + ", biSyncApiDesense onNext reqUuid " + reqUuid + " has EXISTED, skip");
                    return;
                }
                CommunicationObject<String> co = new CommunicationObject<String>();
                co.setSendTimeStamp(System.currentTimeMillis());
                co.setExchangeUuid(reqUuid);
                co.setData("");
                desenseReceivedData.put(reqUuid, co);

                ApiRequest dpRequest = JSONUtil.toBean(dpStr, ApiRequest.class); // 其实也适用的 ApiRequest
                ApiMetadataController.submitTask(dpRequest,"generateDesenseParam"); // 静态线程池成员，不会多实例，也线程安全
                log.info(Thread.currentThread().getId() + ", biSyncApiDesense onNext end, has onNext/onComplete ...." + System.identityHashCode(responseObserver));
            }

            @Override
            public void onError(Throwable t) {
                log.info(Thread.currentThread().getId() + ", biSyncApiDesense onError : " + t.getMessage());
            }

            @Override
            public void onCompleted() {
                log.info(Thread.currentThread().getId() + ", biSyncApiDesense onCompleted ");
            }

        };
        return requestStreamObserver;
    }

    @Override
    public StreamObserver<RPCApiSyncRequest> grpcApiParamsToOmdStream(StreamObserver<RPCApiSyncResponse> responseObserver) {
        StreamObserver<RPCApiSyncRequest> requestStreamObserver = new StreamObserver<RPCApiSyncRequest>() {
            @Override
            public void onNext(RPCApiSyncRequest value) {
                String apiParamStr = value.getApisJson();
                if(apiParamStr == null || apiParamStr.isEmpty()) {
                    return;
                }
                // 回传收到的 requestUuid
                String reqUuid = value.getRequestUuid();
                RPCApiSyncResponse apiResponse = RPCApiSyncResponse.newBuilder().setCode("OK").setRequestUuid(reqUuid).build();
                responseObserver.onNext(apiResponse);
                responseObserver.onCompleted();
                // 构建本地缓存，避免重传对业务的影响
                if(paramsReceivedData.get(reqUuid) != null) { // 假如已存在，返回，不提交到业务端
                    log.info(Thread.currentThread().getId() + ", grpcApiParamsToOmdStream onNext reqUuid " + reqUuid + " has EXISTED, skip");
                    return;
                }
                CommunicationObject<String> co = new CommunicationObject<String>();
                co.setSendTimeStamp(System.currentTimeMillis());
                co.setExchangeUuid(reqUuid);
                co.setData("");
                paramsReceivedData.put(reqUuid, co);

                ApiRequest apiParaRequest = JSONUtil.toBean(apiParamStr, ApiRequest.class);
                ApiMetadataController.submitTask(apiParaRequest, "syncApiParamsToOpenMetadata");
                log.info(Thread.currentThread().getId() + ", grpcApiParamsToOmdStream onNext end, has onNext/onComplete ...." + System.identityHashCode(responseObserver));
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(Thread.currentThread().getId() + ", grpcApiParamsToOmdStream onError: onError, so_request: " + this  + ", so_response: " + responseObserver + ",  message: " + t.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println(Thread.currentThread().getId() + ", grpcApiParamsToOmdStream onCompleted");
            }
        };
        return requestStreamObserver;
    }
}
